First task is not recorded / available per `get_task_stream()`

[amotl]

Describe the issue:

When submitting a task after starting a cluster, it is not available per get_task_stream() inquiry. If we "ping" the cluster first by invoking an initial get_task_stream(), everything is accounted for perfectly afterwards.

Minimal Complete Verifiable Example:

#!/usr/bin/env python
from dask.distributed import Client


def main():
    with Client() as client:

        # FIXME: The program succeeds only when this line is active.
        # tasks = client.get_task_stream()

        # Submit dummy task.
        client.submit(lambda: True).result()

        # Inquire tasks and validate the submitted task has been run.
        tasks = client.get_task_stream()
        if not tasks:
            raise RuntimeError("No tasks have been recorded")
        task = tasks[-1]
        assert task["key"].startswith("lambda"), f"Task has wrong key: {task['key']}"
        assert task["status"] == "OK", f"Task status not OK: {task['status']}"
        print("SUCCESS")


if __name__ == "__main__":
    main()

Anything else we need to know?:

We have been running into this on behalf of those downstream exercises:

• I/O: Add option to submit workloads to remote Dask clusters crate/cratedb-toolkit#788
• Dask Distributed: Basic remoting / serverless function example crate/cratedb-examples#1676

Environment:

• Dask version: 2026.3.0
• Python version: 3.13
• Operating System: macOS / Linux
• Install method (conda, pip, source): OCI / conda / uv

Kudos:

Thank you for conceiving Dask. It's a true gem.