DaskTaskRunner

@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )
)

DaskTaskRunner

cluster_class=...

address="tcp://..."

DaskTaskRunner

cluster

address

cluster_class

cluster_kwargs

adapt_kwargs

PBSCluster

python
from pathlib import Path
from typing import Optional

from prefect import flow, task
from prefect_dask import DaskTaskRunner
from dask_jobqueue import PBSCluster


MAX_WORKERS = 10

pbs_cluster_kwargs = {
    # your PBS settings here
}


def make_dask_task_runner(scheduler_address: Optional[str] = None) -> DaskTaskRunner:
    if scheduler_address:
        return DaskTaskRunner(address=scheduler_address)

    return DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )


@task
def do_work(x: int) -> int:
    return x * 2


@flow(log_prints=True, persist_result=False, retries=1, retry_delay_seconds=20)
def child_flow(
    import_file: str,
    working_dir: Path | None = None,
    overwrite: bool = False,
    scheduler_address: str | None = None,
):
    runner = make_dask_task_runner(scheduler_address)

    with runner:
        futures = [runner.submit(do_work, {"x": i}) for i in range(10)]
        results = [future.result() for future in futures]

    return results

run_deployment

That is much easier if the only reason for the second deployment is reuse.

## If you *must* keep two deployments

Then the pattern is:

1. Master flow creates/owns a Dask cluster
2. Master determines the scheduler address
3. Master starts the child deployment and passes `scheduler_address`
4. Child uses `DaskTaskRunner(address=scheduler_address)`
5. If no address is passed, child falls back to `PBSCluster`

### Conceptually

```python
@flow
def child_flow(scheduler_address: str | None = None):
    runner = (
        DaskTaskRunner(address=scheduler_address)
        if scheduler_address
        else DaskTaskRunner(
            cluster_class=PBSCluster,
            cluster_kwargs=pbs_cluster_kwargs,
            adapt_kwargs={"maximum": MAX_WORKERS},
        )
    )
    ...

@flow(task_runner=...)

scheduler_address

DaskTaskRunner

PBSCluster

prefect deploy

prefect deploy --help

DaskTaskRunner

python
from prefect import flow
from prefect.context import FlowRunContext
from prefect_dask import DaskTaskRunner
from dask_jobqueue import PBSCluster

@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )
)
def master_flow():
    task_runner = FlowRunContext.get().task_runner
    scheduler_address = task_runner.client.scheduler.address

    print(f"Dask scheduler address: {scheduler_address}")
    return scheduler_address

client

DaskTaskRunner

client.scheduler.address

get_dask_client()

python
from prefect import flow, task
from prefect_dask import DaskTaskRunner, get_dask_client
from dask_jobqueue import PBSCluster

@task
def get_scheduler_address():
    with get_dask_client() as client:
        return client.scheduler.address

@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )
)
def master_flow():
    address = get_scheduler_address.submit().result()
    print(f"Dask scheduler address: {address}")
    return address

python
from prefect import flow, task
from prefect.context import FlowRunContext
from prefect_dask import DaskTaskRunner
from dask_jobqueue import PBSCluster

@task
def warmup():
    return "ok"

@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )
)
def master_flow():
    warmup.submit().result()

    task_runner = FlowRunContext.get().task_runner
    scheduler_address = task_runner.client.scheduler.address

    print(f"Dask scheduler address: {scheduler_address}")
    return scheduler_address

scheduler_address

DaskTaskRunner(address=scheduler_address)

master_flow

child_flow

PBSCluster

adapt_kwargs={"minimum": 1, "maximum": MAX_WORKERS}

MAX_WORKERS

python
adapt_kwargs={"maximum": MAX_WORKERS}

minimum=1

minimum=1

DaskTaskRunner

adapt_kwargs={"minimum": 1, "maximum": MAX_WORKERS}

python
@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"minimum": 1, "maximum": MAX_WORKERS},
    ),
    log_prints=True,
    persist_result=False,
    retries=1,
    retry_delay_seconds=20,
)
def master_flow():
    ...

minimum=0

minimum=1

minimum=0

minimum=1

minimum=0

minimum=1

PBSCluster

minimum=0

minimum=1

PBSCluster

DaskTaskRunner

DaskTaskRunner(cluster_class=PBSCluster, ...)

python
@flow(task_runner=DaskTaskRunner(cluster_class=PBSCluster, ...))
def master_flow():
    ...

python
@flow
def master_flow():
    with PBSCluster(...) as cluster:
        cluster.adapt(minimum=1, maximum=MAX_WORKERS)
        with Client(cluster) as client:
            scheduler_address = client.scheduler.address
            ...

python
DaskTaskRunner(address=scheduler_address)

python
from pathlib import Path
from prefect import flow, task
from prefect_dask import DaskTaskRunner
from dask_jobqueue import PBSCluster

MAX_WORKERS = 20

pbs_cluster_kwargs = {
    # queue, project, cores, memory, walltime, etc.
}

@task
def do_work(x: int) -> int:
    return x * 2


@flow(log_prints=True, persist_result=False, retries=1, retry_delay_seconds=20)
def child_flow(
    import_file: str,
    working_dir: Path | None = None,
    overwrite: bool = False,
    scheduler_address: str | None = None,
):
    if scheduler_address:
        runner = DaskTaskRunner(address=scheduler_address)
    else:
        runner = DaskTaskRunner(
            cluster_class=PBSCluster,
            cluster_kwargs=pbs_cluster_kwargs,
            adapt_kwargs={"minimum": 1, "maximum": MAX_WORKERS},
        )

    with runner:
        futures = [runner.submit(do_work, {"x": i}) for i in range(10)]
        results = [future.result() for future in futures]

    return results

---

# Best practice for “manage lifetime carefully”

## Best option
*Master explicitly creates the cluster and does not exit until all dependent child runs are complete.*

That is the safest temporary-cluster pattern.

## Less safe option
Have the master flow use `DaskTaskRunner(cluster_class=...)` and try to share its address with child deployments.

Why it’s less safe:
- cluster startup is tied to the task runner
- cleanup is tied to flow completion
- there is no strong guarantee the cluster survives independently of the master run lifecycle

---

# How to keep master alive until child finishes

If you use `run_deployment(...)`, the master should wait for the child flow run to reach a final state before closing the cluster.

Conceptually:

```python
child = await run_deployment(...)

while not child.state.is_final():
    # sleep, refresh state from API
    ...

run_deployment(...)

minimum=1

minimum=0

python
cluster.adapt(minimum=1, maximum=MAX_WORKERS)

minimum=1

master_flow.py

child_flow.py

with runner:

python
runner = DaskTaskRunner(address=scheduler_address)

DaskTaskRunner

address

python
with DaskTaskRunner(address=scheduler_address) as runner:
    ...

python
DaskTaskRunner(
    cluster_class=PBSCluster,
    cluster_kwargs=...,
    adapt_kwargs=...,
)

with runner:

DaskTaskRunner(address=...)

DaskTaskRunner(cluster_class=...)

DaskTaskRunner(address=scheduler_address)

Client

PBSCluster

with runner:

DaskTaskRunner(address=...)

PBSCluster

address

PBSCluster

DaskTaskRunner(address=...)

master_flow

PBSCluster

distributed.Client

scheduler_address

scheduler_address

child_flow

scheduler_address

DaskTaskRunner(address=scheduler_address)

DaskTaskRunner(cluster_class=PBSCluster, ...)

with runner:

child_flow.py

python
from pathlib import Path

from prefect import flow, task
from prefect_dask import DaskTaskRunner
from dask_jobqueue import PBSCluster

MAX_WORKERS = 20

pbs_cluster_kwargs = {
    # Example fields; replace with your real settings
    # "queue": "myqueue",
    # "account": "myproject",
    # "cores": 4,
    # "memory": "16GB",
    # "walltime": "01:00:00",
    # "job_extra_directives": [...],
}


@task
def do_work(x: int) -> int:
    return x * 2


@flow(
    log_prints=True,
    persist_result=False,
    retries=1,
    retry_delay_seconds=20,
)
def child_flow(
    import_file: str,
    working_dir: Path | None = None,
    overwrite: bool = False,
    scheduler_address: str | None = None,
):
    if scheduler_address:
        print(f"Connecting to existing Dask scheduler at {scheduler_address}")
        runner = DaskTaskRunner(address=scheduler_address)
    else:
        print("No scheduler address provided; creating temporary PBS-backed Dask cluster")
        runner = DaskTaskRunner(
            cluster_class=PBSCluster,
            cluster_kwargs=pbs_cluster_kwargs,
            adapt_kwargs={"minimum": 1, "maximum": MAX_WORKERS},
        )

    with runner:
        futures = [runner.submit(do_work, {"x": i}) for i in range(10)]
        results = [future.result() for future in futures]

    print(f"Child flow results: {results}")
    return results

master_flow.py

---

# Why this is the recommended approach

## Why not let the master use `@flow(task_runner=DaskTaskRunner(cluster_class=PBSCluster, ...))`?

Because then the cluster is owned by the master flow’s task runner lifecycle.

That makes cross-deployment sharing much less predictable:
- cluster startup is lazy
- cleanup happens when the flow/task runner exits
- the child deployment depends on infrastructure owned by another flow’s runner

## Why explicit `PBSCluster` + `Client` is better

It gives you direct control over:
- when the cluster starts
- when adaptation is configured
- how the scheduler address is read
- when the cluster is torn down

That’s the safest pattern when multiple deployed flow runs need to share one temporary cluster.

---

# Why `with runner:` is safe in the child

In the child:

```python
runner = DaskTaskRunner(address=scheduler_address)
with runner:
    ...

python
DaskTaskRunner(
    cluster_class=PBSCluster,
    cluster_kwargs=...,
    adapt_kwargs=...,
)

minimum=1

minimum=0

python
cluster.adapt(minimum=1, maximum=MAX_WORKERS)

minimum=0

prefect deploy

bash
prefect deploy ./child_flow.py:child_flow --name child-deployment
prefect deploy ./master_flow.py:master_flow --name master-deployment

run_deployment(...)

python
name="child-flow/child-deployment"

address

print

python
from pathlib import Path
from typing import Optional

from dask_jobqueue import PBSCluster
from prefect import flow, get_run_logger, task
from prefect_dask import DaskTaskRunner

MAX_WORKERS = 20

pbs_cluster_kwargs: dict = {
    # Replace with your real PBS settings
    # "queue": "myqueue",
    # "account": "myproject",
    # "cores": 4,
    # "memory": "16GB",
    # "walltime": "01:00:00",
    # "job_extra_directives": [...],
}


def build_child_runner(scheduler_address: Optional[str]) -> DaskTaskRunner:
    if scheduler_address:
        return DaskTaskRunner(address=scheduler_address)

    return DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"minimum": 1, "maximum": MAX_WORKERS},
    )


@task
def do_work(x: int) -> int:
    return x * 2


@flow(
    log_prints=False,
    persist_result=False,
    retries=1,
    retry_delay_seconds=20,
)
def child_flow(
    import_file: str,
    working_dir: Optional[Path] = None,
    overwrite: bool = False,
    scheduler_address: Optional[str] = None,
) -> list[int]:
    logger = get_run_logger()

    if not import_file.strip():
        raise ValueError("`import_file` must be a non-empty string.")

    if working_dir is not None and not isinstance(working_dir, Path):
        working_dir = Path(working_dir)

    runner = build_child_runner(scheduler_address)

    if scheduler_address:
        <http://logger.info|logger.info>(
            "Connecting child flow to existing Dask scheduler at %s",
            scheduler_address,
        )
    else:
        <http://logger.info|logger.info>("No scheduler address provided; creating temporary PBS-backed Dask cluster.")

    try:
        with runner:
            futures = [runner.submit(do_work, {"x": i}) for i in range(10)]
            results = [future.result() for future in futures]

        <http://logger.info|logger.info>("Child flow completed successfully with %d results.", len(results))
        return results

    except Exception:
        logger.exception("Child flow failed while executing work on Dask.")
        raise

---

A few notes on these specs:

## 1. Timeout behavior
If you set `child_timeout_seconds`, the master will stop waiting and raise `TimeoutError`. It will still enter `finally` and attempt to close the Dask client and cluster.

## 2. Child standalone behavior
If you run the child deployment directly without `scheduler_address`, it will create its own temporary `PBSCluster` and clean it up when the runner context exits.

## 3. Passing `working_dir`
I convert `working_dir` to `str` in the master when passing deployment parameters because deployment parameters are safest when JSON-serializable. The child converts it back to `Path` if needed.

## 4. Shared cluster safety
When the child uses:

```python
DaskTaskRunner(address=scheduler_address)

prefect.yaml

prefect deploy

PBSCluster

cluster.adapt(minimum=X, maximum=Y)

X

python
cluster.adapt(minimum=1, maximum=MAX_WORKERS)

MAX_WORKERS

minimum=1

minimum=4

minimum=0

DaskTaskRunner(... adapt_kwargs={"maximum": MAX_WORKERS})

python
cluster.adapt(maximum=MAX_WORKERS)

python
cluster.adapt(minimum=0, maximum=MAX_WORKERS)

minimum=0

minimum=1

python
with PBSCluster(...) as cluster:
    with Client(cluster) as client:
        ...

distributed.Client

DaskTaskRunner

distributed.Client(...)

DaskTaskRunner(...)

DaskTaskRunner

PBSCluster

DaskTaskRunner(address=...)

distributed.Client.submit(...)

DaskTaskRunner

adapt(maximum=X)

python
cluster.adapt(maximum=X)

python
cluster.adapt(minimum=1, maximum=X)

python
cluster.scale(2)

PBSCluster

cluster.adapt(minimum=0, maximum=MAX_WORKERS)

scheduler_address

DaskTaskRunner(address=scheduler_address)

DaskTaskRunner(cluster_class=PBSCluster, ...)

.submit()

python
cluster.adapt(minimum=1, maximum=MAX_WORKERS)

python
cluster.scale(2)

python
@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )
)

minimum=X

cluster.scale(...)

python
cluster = PBSCluster(**pbs_cluster_kwargs)
cluster.adapt(minimum=0, maximum=MAX_WORKERS)
client = Client(cluster)

<http://logger.info|logger.info>("Created shared Dask scheduler at %s", client.scheduler.address)
<http://logger.info|logger.info>("Dask dashboard available at %s", client.dashboard_link)

python
cluster = PBSCluster(**pbs_cluster_kwargs)
cluster.scale(2)
client = Client(cluster)

<http://logger.info|logger.info>("Created shared Dask scheduler at %s", client.scheduler.address)
<http://logger.info|logger.info>("Dask dashboard available at %s", client.dashboard_link)

DaskTaskRunner(address=...)

cluster.scale(...)

python
runner = build_child_runner(...)
with runner:
    # do processing here

DaskTaskRunner

python
my_task.submit()

runner

python
@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )
)
def child_flow():
    ...

DaskTaskRunner

task.submit()

with runner:

task.submit()

FlowRunContext

python
@flow
def child_flow():
    runner = DaskTaskRunner(...)
    with runner:
        my_task.submit()

Is there an alternative way to register the cluster that mirrors the behaviour with the decorator (static) approach?

flow.with_options(...)

python
child_flow.with_options(task_runner=...)

python
from pathlib import Path
from typing import Optional

from prefect import flow, task, get_run_logger

@task
def do_work(x: int) -> int:
    return x * 2


@flow(
    log_prints=False,
    persist_result=False,
    retries=1,
    retry_delay_seconds=20,
)
def child_flow(
    import_file: str,
    working_dir: Optional[Path] = None,
    overwrite: bool = False,
    scheduler_address: Optional[str] = None,
) -> list[int]:
    logger = get_run_logger()

    <http://logger.info|logger.info>("Starting child flow")

    futures = [do_work.submit(i) for i in range(10)]
    results = [f.result() for f in futures]

    <http://logger.info|logger.info>("Completed child flow")
    return results

python
from typing import Optional
from dask_jobqueue import PBSCluster
from prefect_dask import DaskTaskRunner

MAX_WORKERS = 20

pbs_cluster_kwargs = {
    # your settings
}

def build_child_runner(scheduler_address: Optional[str]) -> DaskTaskRunner:
    if scheduler_address:
        return DaskTaskRunner(address=scheduler_address)

    return DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )

python
def run_child_flow(
    import_file: str,
    working_dir: Optional[Path] = None,
    overwrite: bool = False,
    scheduler_address: Optional[str] = None,
):
    configured_flow = child_flow.with_options(
        task_runner=build_child_runner(scheduler_address)
    )

    return configured_flow(
        import_file=import_file,
        working_dir=working_dir,
        overwrite=overwrite,
        scheduler_address=scheduler_address,
    )

child_flow

with_options(...)

python
@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )
)
def child_flow_standalone(...):
    ...

address=...

with_options(...)

python
@flow
def child_flow_logic(...):
    futures = [do_work.submit(i) for i in range(10)]
    return [f.result() for f in futures]

python
def child_flow_from_address(scheduler_address: str):
    configured = child_flow_logic.with_options(
        task_runner=DaskTaskRunner(address=scheduler_address)
    )
    return configured

FlowRunContext

python
from prefect import flow, task, get_run_logger
from prefect.context import FlowRunContext
from prefect_dask import DaskTaskRunner
from dask_jobqueue import PBSCluster

@task
def warmup():
    return "ok"

@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    )
)
def child_flow():
    logger = get_run_logger()

    warmup.submit().result()

    task_runner = FlowRunContext.get().task_runner
    scheduler_address = task_runner.client.scheduler.address

    <http://logger.info|logger.info>("Dask scheduler address: %s", scheduler_address)

python
@flow
def child_flow():
    runner = DaskTaskRunner(...)
    with runner:
        my_task.submit()

task.submit()

python
configured_flow = child_flow.with_options(task_runner=DaskTaskRunner(...))
configured_flow(...)

python
@flow(task_runner=DaskTaskRunner(...))
def child_flow():
    ...

child_flow_standalone

cluster_class=PBSCluster

child_flow_shared

address=...

with runner:

prefect-dask

child_flow_standalone

child_flow_shared

child_flow_shared

with runner:

DaskTaskRunner(address=...)

DaskTaskRunner(cluster_class=PBSCluster, ...)

child_flows.py

---

# Critical note about your original design

Your original design is:

- master creates a *temporary* cluster
- master reads that temporary scheduler address
- master launches child deployment
- child connects to *that runtime-created address*

That means the child’s shared-cluster address is *dynamic*, not static.

So the code above with:

```python
task_runner=DaskTaskRunner(address=SHARED_SCHEDULER_ADDRESS)

child_flow_shared

child_flow_shared

python
task_runner=DaskTaskRunner(address="<tcp://stable-hostname:8786>")

with_options(...)

child_shared_logic.py

python
from pathlib import Path
from typing import Optional

from prefect import flow, task, get_run_logger


@task
def do_work(x: int) -> int:
    return x * 2


@flow(
    name="child-flow-logic",
    log_prints=False,
    persist_result=False,
    retries=1,
    retry_delay_seconds=20,
)
def child_flow_logic(
    import_file: str,
    working_dir: Optional[Path] = None,
    overwrite: bool = False,
) -> list[int]:
    logger = get_run_logger()

    if not import_file.strip():
        raise ValueError("`import_file` must be a non-empty string.")

    if working_dir is not None and not isinstance(working_dir, Path):
        working_dir = Path(working_dir)

    <http://logger.info|logger.info>(
        "Running child flow logic with import_file=%s working_dir=%s overwrite=%s",
        import_file,
        working_dir,
        overwrite,
    )

    futures = [do_work.submit(i) for i in range(10)]
    results = [f.result() for f in futures]

    <http://logger.info|logger.info>("Completed child flow logic with %d results.", len(results))
    return results

child_standalone.py

This gives you a deployed flow object named `child_flow_standalone` that behaves like your working decorator-based version.

---

## File: `child_shared.py`

This is the key pattern for dynamic scheduler address from the master.

```python
import os
from pathlib import Path
from typing import Optional

from prefect import flow
from prefect_dask import DaskTaskRunner

from child_shared_logic import child_flow_logic


@flow(name="child-flow-shared-entrypoint")
def child_flow_shared(
    import_file: str,
    working_dir: Optional[Path] = None,
    overwrite: bool = False,
) -> list[int]:
    scheduler_address = os.environ.get("DASK_SCHEDULER_ADDRESS")

    if not scheduler_address:
        raise ValueError(
            "DASK_SCHEDULER_ADDRESS must be set for child_flow_shared."
        )

    configured_flow = child_flow_logic.with_options(
        name="child-flow-shared",
        task_runner=DaskTaskRunner(address=scheduler_address),
    )

    return configured_flow(
        import_file=import_file,
        working_dir=working_dir,
        overwrite=overwrite,
    )

task_runner

PBSCluster

scheduler_address

child_flow_shared

DASK_SCHEDULER_ADDRESS

run_deployment(...)

job_variables

run_deployment

job_variables

python
await run_deployment(
    name="child-flow-shared/child-shared-deployment",
    parameters={
        "import_file": import_file,
        "working_dir": str(working_dir) if working_dir else None,
        "overwrite": overwrite,
    },
    job_variables={
        "env": {
            "DASK_SCHEDULER_ADDRESS": scheduler_address,
        }
    },
)

job_variables={"env": ...}

---

# Final recommendation hierarchy

## *Best if possible*
Use a *stable scheduler address* for the shared cluster.
Then `child_flow_shared` can just use a static decorator with `address=...`.

## *Best for dynamic temporary master-owned clusters*
Use:
- one common logic flow
- one standalone configured entrypoint
- one shared wrapper entrypoint that reads env and calls `with_options(...)`

That is the most realistic way to preserve correct Prefect/Dask behavior while still handling a runtime-generated scheduler address.

## *Avoid*
Do not do this:
```python
@flow
def child_flow():
    runner = DaskTaskRunner(...)
    with runner:
        do_work.submit(...)

master_flow.py

child_shared_logic.py

child_entrypoints.py

python
await run_deployment(
    name="child-flow-shared/child-shared-deployment",
    parameters={
        "import_file": import_file,
        "scheduler_address": scheduler_address,
    },
)

DaskTaskRunner(address=...)

scheduler_address

task.submit()

scheduler_address

run_deployment(..., parameters={"scheduler_address": scheduler_address})

scheduler_address

with_options(task_runner=DaskTaskRunner(address=scheduler_address))

task.submit()

python
@flow
def child_flow():
    runner = DaskTaskRunner(...)
    with runner:
        do_work.submit(...)

child_shared_logic.py

python
from pathlib import Path
from typing import Optional

from prefect import flow, task, get_run_logger


@task
def do_work(x: int) -> int:
    return x * 2


@task
def warmup() -> str:
    return "ok"


@flow(
    name="child-flow-logic",
    log_prints=False,
    persist_result=False,
    retries=1,
    retry_delay_seconds=20,
)
def child_flow_logic(
    import_file: str,
    working_dir: Optional[Path] = None,
    overwrite: bool = False,
) -> list[int]:
    logger = get_run_logger()

    if not import_file or not import_file.strip():
        raise ValueError("`import_file` must be a non-empty string.")

    if working_dir is not None and not isinstance(working_dir, Path):
        working_dir = Path(working_dir)

    <http://logger.info|logger.info>(
        "Running child flow logic with import_file=%s working_dir=%s overwrite=%s",
        import_file,
        str(working_dir) if working_dir else None,
        overwrite,
    )

    # Optional warmup to force Dask runner/client initialization
    warmup.submit().result()

    futures = [do_work.submit(i) for i in range(10)]
    results = [future.result() for future in futures]

    <http://logger.info|logger.info>("Child flow logic completed successfully with %d results.", len(results))
    return results

child_entrypoints.py

master_flow.py

---

# Why this version matches your intent

## Master
The master uses the *working static decorator pattern*:

```python
task_runner=DaskTaskRunner(
    cluster_class=PBSCluster,
    cluster_kwargs=pbs_cluster_kwargs,
    adapt_kwargs={"maximum": MAX_WORKERS},
)

python
scheduler_address: str

python
child_flow_logic.with_options(
    task_runner=DaskTaskRunner(address=scheduler_address)
)

bash
prefect deploy ./child_entrypoints.py:child_flow_standalone --name child-standalone-deployment
prefect deploy ./child_entrypoints.py:child_flow_shared --name child-shared-deployment
prefect deploy ./master_flow.py:master_flow --name master-deployment

prefect deploy

with_options(DaskTaskRunner(address=...))

prefect deploy

prefect.yaml

prefect.yaml

child_flow_standalone

child_flow_shared

master_flow

child_shared_logic.py

child_entrypoints.py

master_flow.py

prefect.yaml

yaml
name: pbs-dask-project
prefect-version: "3"

build: []

push: []

pull: []

deployments:
  - name: child-standalone-deployment
    entrypoint: child_entrypoints.py:child_flow_standalone
    description: Standalone child flow that creates its own temporary PBS-backed Dask cluster.
    work_pool:
      name: my-process-pool
    parameters:
      import_file: ""
      working_dir: null
      overwrite: false

  - name: child-shared-deployment
    entrypoint: child_entrypoints.py:child_flow_shared
    description: Child flow that connects to an existing Dask scheduler passed in at runtime.
    work_pool:
      name: my-process-pool
    parameters:
      import_file: ""
      scheduler_address: ""
      working_dir: null
      overwrite: false

  - name: master-deployment
    entrypoint: master_flow.py:master_flow
    description: Master flow that creates a temporary PBS-backed Dask cluster and triggers the shared child deployment.
    work_pool:
      name: my-process-pool
    parameters:
      import_file: ""
      working_dir: null
      overwrite: false
      child_deployment_name: child-flow-shared/child-shared-deployment
      child_timeout_seconds: null
      poll_interval_seconds: 10

child_deployment_name

master_flow.py

python
child_deployment_name: str = "child-flow-shared/child-shared-deployment"

prefect.yaml

yaml
child_deployment_name: child-flow-shared/child-shared-deployment

child-flow-shared

child-shared-deployment

scheduler_address

child-shared-deployment

yaml
scheduler_address: ""

python
run_deployment(
    ...,
    parameters={
        "scheduler_address": scheduler_address,
    },
)

working_dir

yaml
working_dir: null

bash
prefect deploy

yaml
work_pool:
  name: my-process-pool
  job_variables:
    working_dir: /path/to/project

---

# One important thing to verify

After deploying, confirm the exact deployment names in the UI or with the CLI, because the string used by `run_deployment(name=...)` must match what Prefect registered.

If the flow name / deployment name combination differs, update:

```python
child_deployment_name="child-flow-shared/child-shared-deployment"

prefect deploy --name ...

prefect.yaml

timeout=0

python
async def get_flow_run_result(flow_run_id: UUID) -> object:
    async with get_client() as client:
        flow_run: FlowRun = await client.read_flow_run(flow_run_id)
        return await flow_run.state.result()

state.result()

run_deployment(..., timeout=0)

wait_for_flow_run(...)

asyncio.gather(...)

asyncio.as_completed(...)

wait_for_flow_run(...)

read_flow_run()

files_saved

asyncio.as_completed(...)

python
import asyncio
from uuid import UUID

from prefect.client.orchestration import get_client
from prefect.flow_runs import wait_for_flow_run


async def wait_and_get_flow_run_result(flow_run_id: UUID) -> object:
    async with get_client() as client:
        flow_run = await wait_for_flow_run(
            flow_run_id,
            client=client,
            timeout=None,
            log_states=False,
        )

        return await flow_run.state.result(raise_on_failure=True)

to_process

---

# Why this is better

## Your original loop
This is effectively sequential:

```python
for f_process in to_process:
    flow_run = await run_deployment(...)
    files_saved = await get_flow_run_result(flow_run.id)

asyncio.gather(...)

as_completed(...)

python
results = await asyncio.gather(
    *(wait_and_get_flow_run_result(item["flow_run_id"]) for item in submitted_runs)
)

python
return ["file1", "file2"]

python
@flow(persist_result=True, ...)
def child_flow(...):
    ...

persist_result=False

python
persist_result=False

python
@flow(
    persist_result=True,
    ...
)
def child_flow(...):
    ...

If you want failures returned instead of raised:

```python
async def wait_and_get_flow_run_result(flow_run_id: UUID) -> object:
    async with get_client() as client:
        flow_run = await wait_for_flow_run(
            flow_run_id,
            client=client,
            timeout=None,
            log_states=False,
        )

        return await flow_run.state.result(raise_on_failure=False)

timeout=0

asyncio.as_completed(...)

wait_for_flow_run(...)

read_flow_run(...)

s2_satproc_master

files_saved

while True:

return

raise RuntimeError

raise TimeoutError

python
while True:

python
return

python
raise RuntimeError(...)

python
raise TimeoutError(...)

python
async with get_client() as client:
    while True:

python
flow_run = await client.read_flow_run(flow_run_id)
state = flow_run.state

python
if state is not None and state.type in FINAL_STATES:

COMPLETED

python
if state.type != StateType.COMPLETED:
    raise RuntimeError(...)
return

python
if timeout_seconds is not None:
    elapsed = time.monotonic() - start_time
    if elapsed > timeout_seconds:
        raise TimeoutError(...)

python
await asyncio.sleep(poll_interval_seconds)

while True

python
done = False
while not done:
    check status
    if success:
        done = True
    elif failed:
        raise error
    elif timed out:
        raise timeout
    else:
        sleep

while True

flow_run

None

python
async def wait_for_flow_run(...):
    ...
    while True:
        flow_run = await client.read_flow_run(flow_run_id)
        state = flow_run.state

        if state is not None and state.type in FINAL_STATES:
            if state.type != StateType.COMPLETED:
                raise RuntimeError(...)
            return flow_run

        ...

python
from prefect.flow_runs import wait_for_flow_run

while True

prefect.flow_runs.wait_for_flow_run(...)

asyncio.as_completed(...)

python
if __name__ == "__main__":
    asyncio.run(s2_satproc_master())

python
@flow(
    task_runner=DaskTaskRunner(...)
)
async def s2_satproc_master(...):

asyncio.run(...)

python
asyncio.run(s2_satproc_master())

python
if __name__ == "__main__":
    s2_satproc_master(...)

python
asyncio.run(s2_satproc_master(...))

python
task_runner = flow_ctx.task_runner
scheduler_address = task_runner.client.scheduler.address

run_deployment(...)

@task

DaskTaskRunner

python
task_runner.client

DaskTaskRunner

asyncio.run(...)

python
if __name__ == "__main__":
    asyncio.run(s2_satproc_master())

python
if __name__ == "__main__":
    s2_satproc_master(
        files_list=[...],
        working_dir=Path("/some/path"),
    )

__main__

scheduler_address

run_deployment(...)

python
job_variables={
    "env": {
        "DASK_SCHEDULER_ADDRESS": scheduler_address,
    }
}

scheduler_address

python
main_params = {
    "import_file": f_process.infile,
    "working_dir": str(working_dir),
    "overwrite": overwrite,
    "scheduler_address": scheduler_address,
}

python
flow_run = await run_deployment(
    name=f"sen2-satproc-shared/s2_satproc_shared_{env_state}",
    parameters=main_params,
    timeout=0,
)

with_options(task_runner=DaskTaskRunner(address=scheduler_address))

python
@flow
def child_flow_shared(
    import_file: str,
    scheduler_address: str,
    working_dir: Path | None = None,
    overwrite: bool = False,
):
    configured_flow = child_flow_logic.with_options(
        task_runner=DaskTaskRunner(address=scheduler_address)
    )
    return configured_flow(
        import_file=import_file,
        working_dir=working_dir,
        overwrite=overwrite,
    )

with runner:

task_runner.client

python
from prefect import task

@task
def warmup() -> str:
    return "ok"

python
warmup.submit().result()

python
task_runner = flow_ctx.task_runner
scheduler_address = task_runner.client.scheduler.address

asyncio.as_completed(...)

asyncio.run(...)

DaskTaskRunner(address=...)

asyncio.run(...)

warmup.submit().result()

task_runner.client.scheduler.address

scheduler_address

parameters

job_variables

with_options(task_runner=DaskTaskRunner(address=scheduler_address))

master

child shared

asyncio.run(...)

s2_satproc_master

async def

python
if __name__ == "__main__":
    asyncio.run(
        s2_satproc_master(
            files_list=[...],
            working_dir=Path("/tmp"),
        )
    )

python
RuntimeWarning: coroutine 'run_flow_async' was never awaited

asyncio.run(...)

python
SHARED_SCHEDULER_ADDRESS = os.getenv("DASK_SCHEDULER_ADDRESS")

@flow(
    task_runner=DaskTaskRunner(address=SHARED_SCHEDULER_ADDRESS),
    ...
)
def sen2_satproc_shared(...):
    ...
    return child_flow(...)

python
SHARED_SCHEDULER_ADDRESS = os.getenv("DASK_SCHEDULER_ADDRESS")

python
@flow(
    task_runner=DaskTaskRunner(address=SHARED_SCHEDULER_ADDRESS),
)

python
SHARED_SCHEDULER_ADDRESS

None

python
DaskTaskRunner(address=None)

DASK_SCHEDULER_ADDRESS

python
@flow(
    task_runner=DaskTaskRunner(address=SHARED_SCHEDULER_ADDRESS),
)
def sen2_satproc_shared(...):
    ...
    return child_flow(...)

python
return child_flow(  # now a plain python function with the business logic

child_flow

sen2_satproc_shared

child_flow

python
def child_flow(...):
    # plain python business logic
    futures = [my_task.submit(...)]

python
@flow
def child_flow(...):
    ...

python
SHARED_SCHEDULER_ADDRESS = os.getenv("DASK_SCHEDULER_ADDRESS")

This avoids freezing the scheduler address at import time.

---

# But since you asked about your current exact approach

## Is this valid?

```python
SHARED_SCHEDULER_ADDRESS = os.getenv("DASK_SCHEDULER_ADDRESS")

@flow(task_runner=DaskTaskRunner(address=SHARED_SCHEDULER_ADDRESS))
def sen2_satproc_shared(...):

python
@flow
def sen2_satproc_shared_entrypoint(import_file: str, scheduler_address: str, ...):

python
configured = shared_logic_flow.with_options(
    task_runner=DaskTaskRunner(address=scheduler_address)
)
return configured(...)

python
SHARED_SCHEDULER_ADDRESS = os.getenv("DASK_SCHEDULER_ADDRESS")

scheduler_address

with_options(task_runner=DaskTaskRunner(address=scheduler_address))

asyncio.run(...)

async def

python
Preparing shared child flow using external Dask scheduler at tcp://...

run_deployment(...)

job_variables.env.DASK_SCHEDULER_ADDRESS

run_deployment(...)

job_variables

python
import socket
from prefect import task
from prefect_dask import get_dask_client

@task
def debug_dask_execution() -> str:
    host = socket.gethostname()

    with get_dask_client() as client:
        scheduler = client.scheduler.address
        workers = list(client.scheduler_info()["workers"].keys())

    return f"host={host}, scheduler={scheduler}, workers={workers}"

python
info = debug_dask_execution.submit().result()
<http://logger.info|logger.info>("Dask execution diagnostic: %s", info)

.submit()

python
@flow(task_runner=DaskTaskRunner(...))
def sen2_satproc_shared(...):
    result = expensive_python_function(...)
    return result

expensive_python_function(...)

python
@task
def expensive_python_function(...):
    ...

@flow(task_runner=DaskTaskRunner(...))
def sen2_satproc_shared(...):
    future = expensive_python_function.submit(...)
    return future.result()

maybe env passing to child won’t work

run_deployment(..., job_variables={"env": ...})

.submit()

python
import os
import socket
from prefect import task
from prefect_dask import get_dask_client

@task
def debug_cluster() -> dict:
    with get_dask_client() as client:
        info = client.scheduler_info()

    return {
        "hostname": socket.gethostname(),
        "pid": os.getpid(),
        "scheduler": info["address"],
        "workers": list(info["workers"].keys()),
    }

python
diag = debug_cluster.submit().result()
<http://logger.info|logger.info>("Cluster diagnostics: %s", diag)

run_deployment(..., job_variables={"env": ...})

debug_cluster()

python
diag = debug_cluster.submit().result()

python
flow_run = await run_deployment(...)
files_saved = await get_flow_run_result(flow_run.id)

asyncio.as_completed(...)

python
  Preparing shared child flow using external Dask scheduler at tcp://...

python
  diag = debug_cluster.submit().result()

debug_cluster.submit().result()

python
diag = debug_cluster.submit().result()

get_dask_client()

python
from prefect_dask import get_dask_client

@flow(...)
def sen2_satproc_shared(...):
    logger = get_run_logger()

    with get_dask_client() as client:
        info = client.scheduler_info()
        <http://logger.info|logger.info>("Scheduler info in flow context: %s", {
            "address": info["address"],
            "workers": list(info["workers"].keys()),
        })

    diag = debug_cluster.submit().result()

debug_cluster.submit().result()

python
with get_dask_client() as client:
    info = client.scheduler_info()
    <http://logger.info|logger.info>("Connected workers: %s", list(info["workers"].keys()))

python
adapt_kwargs={"maximum": MAX_WORKERS}

DaskTaskRunner(cluster_class=PBSCluster, ...)

cluster.scale(...)

python
with PBSCluster(...) as cluster:
    cluster.scale(2)
    with Client(cluster) as client:
        scheduler_address = client.scheduler.address
        ...

asyncio.as_completed(...)

debug_cluster.submit().result()

python
with get_dask_client() as client:
    info = client.scheduler_info()
    <http://logger.info|logger.info>(
        "Scheduler connected in flow context: address=%s workers=%s",
        info["address"],
        list(info["workers"].keys()),
    )

debug_cluster.submit().result()

as_completed(...)

asyncio.gather

get_dask_client()

python
with get_dask_client() as client:

None

python
client.scheduler_info()

python
@flow(
    task_runner=DaskTaskRunner(address=SHARED_SCHEDULER_ADDRESS),
    ...
)

python
SHARED_SCHEDULER_ADDRESS = os.getenv("DASK_SCHEDULER_ADDRESS")

@flow(
    task_runner=DaskTaskRunner(address=SHARED_SCHEDULER_ADDRESS),
)
def sen2_satproc_shared(...):

DaskTaskRunner

os.getenv(...)

task_runner

python
SHARED_SCHEDULER_ADDRESS = os.getenv("DASK_SCHEDULER_ADDRESS")
@flow(task_runner=DaskTaskRunner(address=SHARED_SCHEDULER_ADDRESS))

with_options(...)

DaskTaskRunner(address=...)

child_shared_logic.py

python
from pathlib import Path
from typing import Optional

from prefect import task, get_run_logger

@task
def debug_cluster() -> dict:
    import os
    import socket
    from prefect_dask import get_dask_client

    with get_dask_client() as client:
        info = client.scheduler_info()

    return {
        "hostname": socket.gethostname(),
        "pid": os.getpid(),
        "scheduler": info["address"],
        "workers": list(info["workers"].keys()),
    }


def child_flow_logic(
    import_file: str,
    working_dir: Optional[Path] = None,
    overwrite: bool = False,
) -> list[str]:
    logger = get_run_logger()

    diag = debug_cluster.submit().result()
    <http://logger.info|logger.info>("Cluster diagnostics: %s", diag)

    # your real logic here
    return []

child_entrypoints.py

---

# Why this is better

Because now the Dask runner is created from:

```python
scheduler_address = os.getenv("DASK_SCHEDULER_ADDRESS")

python
_sen2_satproc_shared_logic.with_options(
    task_runner=DaskTaskRunner(address=scheduler_address)
)

get_dask_client()

None

AttributeError: 'NoneType' object has no attribute 'scheduler_info'

prefect.yaml

async def

def

DaskTaskRunner

prefect-dask

asyncio.run(...)

python
INFO | prefect.task_runner.dask - Creating a new Dask cluster with `dask_jobqueue.pbs.PBSCluster`

.submit()

python
@flow(...)
async def s2_satproc_master(...):

python
@task
def warmup():
    return "ok"

python
from prefect import flow, task
from prefect_dask import DaskTaskRunner
from dask_jobqueue import PBSCluster

pbs_cluster_kwargs = {
    # your known-good settings
}

@task
def hello():
    import socket
    return socket.gethostname()

@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": 2},
    )
)
def test_flow():
    return hello.submit().result()

if __name__ == "__main__":
    print(test_flow())

python
@task
async def hello_async():
    import socket
    return socket.gethostname()

python
@flow(...)
def my_flow():
    expensive_python_function()

python
@task
def expensive_python_function():
    ...

@flow(...)
def my_flow():
    expensive_python_function.submit()

.bashrc

.bash_profile

conda activate

bash
[[ $- == *i* ]] || return

/tmp

$TMPDIR

python

prefect

prefect-dask

dask

distributed

dask-jobqueue

DaskTaskRunner

PBSCluster

PBSCluster

python
from dask_jobqueue import PBSCluster
from distributed import Client

cluster = PBSCluster(
    # use the same settings as your Prefect flow
    # queue=...,
    # account=...,
    # cores=...,
    # memory=...,
    # walltime=...,
)

print("Cluster created")
print(cluster.job_script())

cluster.scale(1)

client = Client(cluster)
print("Dashboard:", client.dashboard_link)
print("Scheduler:", client.scheduler.address)

print(client.scheduler_info())
input("Press Enter to close...")

cluster.scale(1)

bash
qstat -u <your_user>

.bashrc

bash
timeout 5 /bin/bash -i -c 'echo OK'

bash
bash -lc 'python -V'

.bashrc

bash
[[ $- == *i* ]] || return

bash
echo $TMPDIR
ls -ld /tmp
ls -ld "$HOME"

bash
touch /tmp/test_file.$$ && rm /tmp/test_file.$$
mkdir -p "$HOME/tmp_test_dir" && rmdir "$HOME/tmp_test_dir"

PBSCluster

/tmp

pbs_cluster_kwargs

bash
#!/bin/bash
#PBS -N test_python_env
#PBS -o /path/to/writeable/stdout.log
#PBS -e /path/to/writeable/stderr.log

which python
python -V
python -c "import prefect, distributed, dask_jobqueue; print('ok')"

PBSCluster

python
print(cluster.job_script())

.bashrc

.bash_profile

bash
timeout 5 bash -lc 'python -V'
timeout 5 bash -lc 'which python'
timeout 5 bash -lc 'python -c "import distributed, dask_jobqueue; print(\"ok\")"'

bash
python -c "import tempfile; print(tempfile.gettempdir())"
touch /tmp/pbs_test.$$ && rm /tmp/pbs_test.$$

bash
qstat -u $USER

bash
env | sort
which python
python -V
python -c "import prefect, distributed, dask_jobqueue; print(prefect.__version__)"

DaskTaskRunner

PBSCluster + Client

python
from dask_jobqueue import PBSCluster
from distributed import Client

cluster = PBSCluster(
    # same kwargs you use in Prefect
)

print("Created cluster")
print(cluster.job_script())

cluster.scale(1)
print("Scaled to 1 worker")

client = Client(cluster)
print("Connected client")
print("Scheduler:", client.scheduler.address)
print("Info:", client.scheduler_info())

pbs_cluster_kwargs

processing: 0

no-worker: 1

no-worker: 1

.bashrc

.bash_profile

python
info = client.scheduler_info()
print(info["workers"].keys())

bash
qstat -u $USER

ModuleNotFoundError

Permission denied

No such file or directory

OSError

/tmp

bash
python -c "import tempfile; print(tempfile.gettempdir())"
touch /tmp/test.$$ && rm /tmp/test.$$

bash
timeout 5 bash -lc 'python -V'
timeout 5 bash -lc 'python -c "import distributed, dask_jobqueue; print(\"ok\")"'

no-worker: 1

asyncio.as_completed

run_deployment

PBSCluster

DaskTaskRunner

yaml
pull:
  - prefect.deployments.steps.git_clone:
      repository: <https://gitlab.com/my_project.git>
      branch: my-feat-branch
      credentials: "{{ prefect.blocks.gitlab-credentials.my-token }}"

git_clone

prefect.deployments.steps.git_clone

working_dir

git_clone

working_dir

prefect.deployments.steps.git_clone

git_clone

yaml
job_variables:
  working_dir: /my_scratch/dir/prefect-run-<unique-id>

git_clone

yaml
pull:
  - prefect.deployments.steps.git_clone:
      repository: <https://gitlab.com/my_project.git>
      branch: my-feat-branch
      credentials: "{{ prefect.blocks.gitlab-credentials.my-token }}"

yaml
pull: []

pull

working_dir

yaml
work_pool:
  name: athena_dev
  job_variables:
    working_dir: /my_scratch/dir/my_project_checkout

yaml
entrypoint: flows/child_flow.py:sen2_satproc_shared

bash
cd /my_scratch/dir/my_project_checkout
git fetch
git checkout my-feat-branch
git pull

python
for f_process in to_process:
    flow_run = await run_deployment(...)

git_clone

yaml
name: overnight_prefect
prefect-version: 3.6.22
build: null
push: null
pull: []

deployments:
  - name: s2_satproc_shared_dev
    description: "Creating sentinel2 QVF downstream products - shared cluster deployment."
    entrypoint: flows/child_flow.py:sen2_satproc_shared
    parameters:
      import_file: "test.img"
      working_dir: "/my_scratch/dir"
      overwrite: false
    work_pool:
      name: athena_dev
      work_queue_name: null
      job_variables:
        working_dir: "/my_scratch/dir/my_project_checkout"
        env:
          RSS_DEFAULT_FILEMODE: "644"
          PREFECT_PYTHON_CMD: "x/x/xx/python"
          SATPROC_SHARED_DIR: "{{ prefect.variables.satproc_workdir_dev }}"
          DASK_SCHEDULER_ADDRESS: null
          GIT_TERMINAL_PROMPT: "0"

/my_scratch/dir/my_project_checkout

flow.from_source(...)

git_clone

git_clone

prefect.yaml

working_dir

git_clone

pull

working_dir

working_dir

text
/my_scratch/prefect-code/
  satproc-main/
  satproc-my-feat-branch/

bash
cd /my_scratch/prefect-code
git clone <https://gitlab.com/my_project.git> satproc-main
cd satproc-main
git checkout main

git clone <https://gitlab.com/my_project.git> satproc-my-feat-branch
cd satproc-my-feat-branch
git checkout my-feat-branch

/my_scratch/prefect-code/satproc-dev

/my_scratch/prefect-code/satproc-my-feat-branch

yaml
job_variables:
  working_dir: /my_scratch/prefect-code/satproc-my-feat-branch

git_clone(branch=...)

working_dir

prefect.yaml

job_variables.working_dir

prefect.yaml

## Why this works
- no `git_clone`
- both master and child resolve entrypoints from the same local branch checkout
- no clone race
- branch selection is controlled by `working_dir`

---

# Option B: Separate YAMLs for master and child

If you want separate deployment files, that’s fine too.

The key is the same: *both point to a controlled local checkout path*.

## `prefect.master.yaml`

```yaml
name: overnight_prefect_master
prefect-version: 3.6.22

build: []
push: []
pull: []

deployments:
  - name: s2_satproc_master_dev
    entrypoint: flows/master_flow.py:s2_satproc_master
    parameters:
      files_list: []
      working_dir: "/my_scratch/data"
      overwrite: false
      env_state: "dev"
    work_pool:
      name: athena_dev
      job_variables:
        working_dir: /my_scratch/prefect-code/satproc-my-feat-branch
        env:
          RSS_DEFAULT_FILEMODE: "644"
          PREFECT_PYTHON_CMD: "/x/x/xx/python"
          SATPROC_SHARED_DIR: "{{ prefect.variables.satproc_workdir_dev }}"

prefect.child.yaml

yaml
name: overnight_prefect_child
prefect-version: 3.6.22

build: []
push: []
pull: []

deployments:
  - name: s2_satproc_shared_dev
    entrypoint: flows/child_flow.py:sen2_satproc_shared
    parameters:
      import_file: "test.img"
      working_dir: "/my_scratch/data"
      overwrite: false
    work_pool:
      name: athena_dev
      job_variables:
        working_dir: /my_scratch/prefect-code/satproc-my-feat-branch
        env:
          RSS_DEFAULT_FILEMODE: "644"
          PREFECT_PYTHON_CMD: "/x/x/xx/python"
          SATPROC_SHARED_DIR: "{{ prefect.variables.satproc_workdir_dev }}"
          DASK_SCHEDULER_ADDRESS: null

pull: []

having

pull: []

effectively defaults to

set_working_directory

job_variables.working_dir

text
/my_scratch/prefect-code/
  satproc-main/
  satproc-dev/
  satproc-my-feat-branch/

yaml
job_variables:
  working_dir: /my_scratch/prefect-code/satproc-my-feat-branch

text
/my_scratch/prefect-code/current -> /my_scratch/prefect-code/satproc-my-feat-branch

yaml
job_variables:
  working_dir: /my_scratch/prefect-code/current

bash
prefect worker start --pool athena_dev

prefect worker start

job_variables

yaml
job_variables:
  working_dir: /my_scratch/prefect-code/satproc-my-feat-branch

python
await run_deployment(
    name=f"sen2-satproc-shared/s2_satproc_shared_{env_state}",
    ...
)

working_dir

master_flow.py

child_flow.py

text
/my_scratch/prefect-code/satproc-my-feat-branch/
    prefect.yaml
    flows/
      master_flow.py
      child_flow.py

yaml
job_variables:
  working_dir: /my_scratch/prefect-code/satproc-my-feat-branch

bash
cd /my_scratch/prefect-code/satproc-my-feat-branch
git fetch
git checkout my-feat-branch
git pull

yaml
job_variables:
  working_dir: "{{ prefect.variables.satproc_code_dir_dev }}"

/my_scratch/prefect-code/satproc-main

/my_scratch/prefect-code/satproc-my-feat-branch

pull: []

git_clone

job_variables.working_dir

yaml
name: overnight_prefect
prefect-version: 3.6.22

build: []
push: []
pull: []

deployments:
  - name: s2_satproc_master_dev
    entrypoint: flows/master_flow.py:s2_satproc_master
    parameters:
      files_list: []
      working_dir: "/my_scratch/data"
      overwrite: false
      env_state: "dev"
    work_pool:
      name: athena_dev
      job_variables:
        working_dir: "{{ prefect.variables.satproc_code_checkout_dev }}"
        env:
          RSS_DEFAULT_FILEMODE: "644"
          PREFECT_PYTHON_CMD: "/x/x/xx/python"
          SATPROC_SHARED_DIR: "{{ prefect.variables.satproc_workdir_dev }}"

  - name: s2_satproc_shared_dev
    entrypoint: flows/child_flow.py:sen2_satproc_shared
    parameters:
      import_file: "test.img"
      working_dir: "/my_scratch/data"
      overwrite: false
    work_pool:
      name: athena_dev
      job_variables:
        working_dir: "{{ prefect.variables.satproc_code_checkout_dev }}"
        env:
          RSS_DEFAULT_FILEMODE: "644"
          PREFECT_PYTHON_CMD: "/x/x/xx/python"
          SATPROC_SHARED_DIR: "{{ prefect.variables.satproc_workdir_dev }}"
          DASK_SCHEDULER_ADDRESS: null

prefect.variables.satproc_code_checkout_dev

text
/my_scratch/prefect-code/satproc-my-feat-branch

git_clone

shared_local_clone

yaml
pull:
  - prefect.deployments.steps.set_working_directory:
      directory: "{{ prefect.variables.shared_local_clone }}"

shared_local_clone

shared_local_clone=/path/clone_A

shared_local_clone=/path/clone_B

job_variables.working_dir

/scratch/prefect-code/run-<flow_run_id>

python
job_variables={
    "working_dir": local_clone_path,
    "env": {
        "DASK_SCHEDULER_ADDRESS": scheduler_address,
    },
}

git_clone

set_working_directory

working_dir

yaml
name: overnight_prefect
prefect-version: 3.6.22
build: []
push: []
pull: []

deployments:
  - name: s2_satproc_shared_dev
    description: "Creating sentinel2 QVF downstream products - shared cluster deployment."
    entrypoint: flows/child_flow.py:sen2_satproc_shared
    parameters:
      import_file: "test.img"
      working_dir: "/my_scratch/dir"
      overwrite: false
    work_pool:
      name: athena_dev
      job_variables:
        working_dir: /tmp/default-placeholder
        env:
          RSS_DEFAULT_FILEMODE: "644"
          PREFECT_PYTHON_CMD: "x/x/xx/python"
          SATPROC_SHARED_DIR: "{{ prefect.variables.satproc_workdir_dev }}"
          DASK_SCHEDULER_ADDRESS: null
          GIT_TERMINAL_PROMPT: "0"

pull: []

git_clone

shared_local_clone

working_dir

python
from pathlib import Path
import subprocess
import uuid

def clone_repo(branch: str, base_dir: Path) -> Path:
    clone_dir = base_dir / f"prefect-run-{uuid.uuid4().hex}"

    subprocess.run(
        [
            "git",
            "clone",
            "--branch",
            branch,
            "<https://gitlab.com/my_project.git>",
            str(clone_dir),
        ],
        check=True,
    )

    return clone_dir

python
local_clone = clone_repo(branch=my_branch_name, base_dir=Path("/my_scratch/prefect-code"))
<http://logger.info|logger.info>("Prepared local clone at %s", local_clone)

That is the key change.

---

# Why `job_variables.working_dir` is the right place

Because for a process worker, the worker job/run process can be told to execute from that working directory for that specific run.

That avoids needing:
- a pull step
- a shared variable
- a checkout preinstalled manually

It also preserves your branch-selection idea:
- branch becomes a master flow parameter
- master clones that branch once
- all children for that master run use that clone

---

# Strong caveat

This is better than per-child `git_clone`, but it still has a tradeoff:

## You are now doing Git clone *from inside the master flow runtime*
That means:
- the master run host/process must have Git/network credentials
- the clone location must be visible to the child worker processes
- cleanup becomes your responsibility

So it’s operationally more complex than a fixed local checkout, but still much safer than concurrent child clones.

---

# Very important filesystem requirement

This only works if the clone path created by the master is visible to the child deployment runtime.

For a process worker, that usually means:
- master and child runs are on the same machine or shared filesystem
- the path is accessible to the worker process that runs the child

If the child process runs elsewhere without shared visibility, this will fail.

Given your HPC/shared filesystem setup, it sounds plausible that this is okay.

---

# Cleanup strategy

Since you’re creating per-master-run clones, you should clean them up eventually.

You have two options:

## Option A: cleanup in master `finally`
After all children complete:

```python
shutil.rmtree(local_clone, ignore_errors=True)

---

# Best YAML for this pattern

## Child deployment YAML

```yaml
name: overnight_prefect
prefect-version: 3.6.22
build: []
push: []
pull: []

deployments:
  - name: s2_satproc_shared_dev
    description: "Creating sentinel2 QVF downstream products - shared cluster deployment."
    entrypoint: flows/child_flow.py:sen2_satproc_shared
    parameters:
      import_file: "test.img"
      working_dir: "/my_scratch/dir"
      overwrite: false
    work_pool:
      name: athena_dev
      job_variables:
        working_dir: /tmp/prefect-placeholder
        env:
          RSS_DEFAULT_FILEMODE: "644"
          PREFECT_PYTHON_CMD: "x/x/xx/python"
          SATPROC_SHARED_DIR: "{{ prefect.variables.satproc_workdir_dev }}"
          DASK_SCHEDULER_ADDRESS: null
          GIT_TERMINAL_PROMPT: "0"

run_deployment(..., job_variables=...)

One of the nice things about

git_clone

is the ability to grab code from a feature branch. How can I achieve something similar?

my_branch_name

master_flow

job_variables.working_dir

job_variables.working_dir

job_variables.env

clone_repo

git_clone

python
cred = GitLabCredentials.load(CRED_BLOCK_NAME)
code_dir = await git_clone(
    GITLAB_REPO, branch=repo_branch, access_token=cred
)

python
TypeError: cannot pickle 'coroutine' object

GitLabCredentials.load(...)

prefect.deployments.steps.git_clone

yaml
pull:
  - prefect.deployments.steps.git_clone:
      repository: ...
      branch: ...
      credentials: "{{ prefect.blocks.gitlab-credentials.my-token }}"

GitLabCredentials

git_clone

prefect.deployments.steps.pull.git_clone

subprocess.run(["git", "clone", ...])

prefect.deployments.steps.git_clone

python
from pathlib import Path
import subprocess
import uuid


def clone_repo(repository: str, branch: str, base_dir: Path) -> Path:
    clone_dir = base_dir / f"repo-{uuid.uuid4().hex}"

    subprocess.run(
        [
            "git",
            "clone",
            "--branch",
            branch,
            repository,
            str(clone_dir),
        ],
        check=True,
    )

    return clone_dir

python
<http://logger.info|logger.info>("Cloning %s:%s ...", GITLAB_REPO, repo_branch)
code_dir = clone_repo(GITLAB_REPO, repo_branch, Path("/my_scratch/prefect-code"))
<http://logger.info|logger.info>("Cloned into %s", code_dir)

python
def build_gitlab_url_with_token(repo_url: str, token: str) -> str:
    return repo_url.replace("https://", f"<https://oauth2:{token}@>")

python
token = GitLabCredentials.load(CRED_BLOCK_NAME).token.get_secret_value()
repo_url = build_gitlab_url_with_token(GITLAB_REPO, token)

load()

python
cred = GitLabCredentials.load(CRED_BLOCK_NAME)

load()

cred

python
TypeError: cannot pickle 'coroutine' object

python
cred = await GitLabCredentials.load(CRED_BLOCK_NAME)

git_clone

GitLabCredentials.load(...)

prefect.deployments.steps.pull.git_clone

GitLabCredentials

job_variables.working_dir

python
from pathlib import Path
import subprocess
import uuid

from prefect_gitlab import GitLabCredentials


def clone_repo_with_token(repo_url: str, branch: str, token: str, base_dir: Path) -> Path:
    clone_dir = base_dir / f"repo-{uuid.uuid4().hex}"
    authed_url = repo_url.replace("https://", f"<https://oauth2:{token}@>")

    subprocess.run(
        [
            "git",
            "clone",
            "--branch",
            branch,
            authed_url,
            str(clone_dir),
        ],
        check=True,
    )

    return clone_dir

python
cred = await GitLabCredentials.load(CRED_BLOCK_NAME)
token = cred.token.get_secret_value()

code_dir = clone_repo_with_token(
    repo_url=GITLAB_REPO,
    branch=repo_branch,
    token=token,
    base_dir=Path("/my_scratch/prefect-code"),
)

prefect.deployments.steps.pull.git_clone

clone_repo

GitLabCredentials

working_dir

git_clone

prefect.yaml

job_variables.working_dir

working_dir

job_variables.working_dir

run_deployment(...)

flowrun.id

python
job_variables={
    "working_dir": working_dir / flowrun.id,
}

flowrun.id

run_deployment(...)

uuid.uuid4()

python
import uuid

clone_dir = working_dir / f"child-run-{uuid.uuid4().hex}"

python
flow_run = await run_deployment(
    ...,
    job_variables={
        "working_dir": str(clone_dir),
        "env": {
            "DASK_SCHEDULER_ADDRESS": scheduler_address,
        },
    },
)

working_dir

I think this is the default behaviour when using a process worker, the worker runs the deployment git_clone step using the work pool’s

working_dir

value.

yaml
name: overnight_prefect
prefect-version: 3.6.22
build: []
push: []
pull:
  - prefect.deployments.steps.git_clone:
      repository: <https://gitlab.com/my_project.git>
      branch: my-feat-branch
      credentials: "{{ prefect.blocks.gitlab-credentials.my-token }}"

deployments:
  - name: s2_satproc_shared_dev
    description: "Creating sentinel2 QVF downstream products - shared cluster deployment."
    entrypoint: flows/child_flow.py:sen2_satproc_shared
    parameters:
      import_file: "test.img"
      working_dir: "/my_scratch/dir"
      overwrite: false
    work_pool:
      name: athena_dev
      job_variables:
        working_dir: /tmp/default-placeholder
        env:
          RSS_DEFAULT_FILEMODE: "644"
          PREFECT_PYTHON_CMD: "x/x/xx/python"
          SATPROC_SHARED_DIR: "{{ prefect.variables.satproc_workdir_dev }}"
          DASK_SCHEDULER_ADDRESS: null
          GIT_TERMINAL_PROMPT: "0"

working_dir

---

# What I would expect this to do

For each child flow run:

1. process worker receives the run
2. worker uses that run’s `job_variables.working_dir`
3. pull step executes in that isolated working directory
4. `git_clone` clones repo there
5. child run uses that isolated checkout
6. no clone race with other child runs

That is the desired behavior.

---

# Why this is better than your earlier shared-dir design

Because before, all child runs were effectively competing for the same clone location.

Now each run gets:
- its own working directory
- its own code checkout
- no shared mutation of the same filesystem path

That is the correct concurrency fix.

---

# Two practical concerns

## 1. Cleanup
If each child run clones into a unique directory, these directories will accumulate.

You’ll probably want:
- a cleanup task
- retention cleanup cron/job
- or a parent/master cleanup after child completion if feasible

For example, structure them under:

```text
/my_scratch/dir/prefect-clones/

yaml
pull:
  - prefect.deployments.steps.git_clone:
      repository: ...
      branch: my-feat-branch

working_dir

git_clone

git_clone

job_variables.working_dir

str(...)

python
import uuid

clone_dir = working_dir / "prefect-clones" / f"{uuid.uuid4().hex}"

flowrun.id

flow_run

job_variables.working_dir

run_deployment(...)

git_clone

working_dir

run

sen2_satproc_standalone

sen2_satproc(...)

run_sfcref_tasks(...)

.submit()

run

sat_proc_tasks

python
from prefect import task

@task(name="run-dem")
def run_dem(...):
    ...

name=

python
def run_sfcref_tasks(...): ...
def sen2_satproc(...): ...

python
@task(name="dem-generation")
def run_dem(...):
    ...

SatProcTasks

python
@flow(name="surface-reflectance-phase")
def run_sfcref_tasks(...):
    dem = sat_proc_tasks.run_dem.submit(...)
    ...
    return ...

python
@task(name="run-dem")
def run_dem(templatefile: str, outfile: str, **kwargs):
    ...

run_sfcref_tasks()

python
from prefect import task

@task(name="run-dem")
def run_dem(templatefile: str, outfile: str, **kwargs) -> str:
    ...

@task(name="run-angles")
def run_angles(...) -> str:
    ...

python
from prefect import flow
from prefect.futures import PrefectFuture

@flow(name="run-sfcref-phase")
def run_sfcref_tasks(
    qvf_stage: str,
    sen2_files: Sen2ProcessFiles,
    sat_proc_tasks: SatProcTasks,
    angles: PrefectFuture[str],
    **kwargs: object,
) -> PrefectFuture[str]:
    dem = sat_proc_tasks.run_dem.submit(
        templatefile=toa_infile,
        outfile=sen2_proc.dem,
        **kwargs,
    )
    ...
    return dem

python
def sen2_satproc(
    import_file: str,
    working_dir: Path | None = None,
    overwrite: bool = False,
) -> list[str]:
    ...
    result = run_sfcref_tasks(
        qvf_stage=qvf_stage,
        sen2_files=sen2_files,
        sat_proc_tasks=sat_proc_tasks,
        angles=angles,
        **kwargs,
    )
    ...
    return files_saved

python
@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    ),
    log_prints=False,
    persist_result=False,
    retries=1,
    retry_delay_seconds=20,
)
def sen2_satproc_standalone(
    import_file: str,
    working_dir: Optional[Path] = None,
    overwrite: bool = False,
) -> list[str]:
    return sen2_satproc(
        import_file=import_file,
        working_dir=working_dir,
        overwrite=overwrite,
    )

name=...

run_sfcref_tasks

run-dem

run-angles

run-toa

run-sfcref

run

prefect-dask

asyncio

asyncio.as_completed(...)

python
submitted child deployments
    ↓
wait for child completion as they finish
    ↓
schedule dependent async follow-up immediately
    ↓
wait for all follow-up tasks

resolve_flow_run(run_info)

process_child_result(r_d)

python
pending_children = [asyncio.create_task(resolve_flow_run(run_info)) for run_info in submitted_runs]
pending_followups = []

for completed in asyncio.as_completed(pending_children):
    r_d = await completed

    # schedule downstream async work immediately
    followup_task = asyncio.create_task(process_child_result(r_d))
    pending_followups.append(followup_task)

# after all children have finished, wait for all follow-up tasks
followup_results = await asyncio.gather(*pending_followups, return_exceptions=True)

python
async def resolve_flow_run(run_info: dict) -> dict:
    result = await wait_and_get_flow_run_result(run_info["flow_run_id"])
    return {
        "flow_run_id": run_info["flow_run_id"],
        "import_file": run_info["import_file"],
        "result": result,
    }

python
async def process_child_output(r_d: dict) -> dict:
    result = r_d["result"]

    if isinstance(result, Exception):
        return {
            "flow_run_id": r_d["flow_run_id"],
            "import_file": r_d["import_file"],
            "status": "failed-upstream",
            "result": result,
        }

    # Do more async work here
    await asyncio.sleep(0)  # placeholder

    return {
        "flow_run_id": r_d["flow_run_id"],
        "import_file": r_d["import_file"],
        "status": "processed",
        "result": result,
    }

python
pending_children = [asyncio.create_task(resolve_flow_run(run_info)) for run_info in submitted_runs]
pending_followups = []

files_saved: list[str | object] = []
failures: list[Exception] = []

for completed in asyncio.as_completed(pending_children):
    r_d = await completed

    <http://logger.info|logger.info>(
        "Child flow run %s finished for %s.",
        r_d["flow_run_id"],
        r_d["import_file"],
    )

    result = r_d["result"]
    if not isinstance(result, Exception):
        <http://logger.info|logger.info>("files saved: result=%s", result)
        files_saved.extend(result)
    else:
        <http://logger.info|logger.info>("%s - %s failed with error:", r_d["import_file"], r_d["flow_run_id"])
        logger.warning(result)
        failures.append(result)

    pending_followups.append(asyncio.create_task(process_child_output(r_d)))

followup_results = await asyncio.gather(*pending_followups, return_exceptions=True)

python
resolved = []
for completed in asyncio.as_completed(pending_children):
    r_d = await completed
    resolved.append(r_d)

final_result = await run_next_async_stage(resolved)

python
semaphore = asyncio.Semaphore(5)

async def process_child_output_limited(r_d: dict) -> dict:
    async with semaphore:
        return await process_child_output(r_d)

python
pending_followups.append(
    asyncio.create_task(process_child_output_limited(r_d))
)

run_deployment(..., timeout=0)

as_completed(...)

gather(...)

asyncio.as_completed(...)

submitted_runs

python
pending_children = [asyncio.create_task(resolve_flow_run(run_info)) for run_info in submitted_runs]
pending_next_stage = []

for completed in asyncio.as_completed(pending_children):
    r_d = await completed

    # handle/log child result
    ...

    # launch next async stage for this completed child
    pending_next_stage.append(
        asyncio.create_task(next_async_stage(r_d))
    )

next_stage_results = await asyncio.gather(*pending_next_stage, return_exceptions=True)

python
from dataclasses import dataclass
from uuid import UUID

@dataclass
class ChildRunResult:
    flow_run_id: UUID
    import_file: str
    result: object

run_deployment()

with ThreadPoolTaskRunner(...)

run_deployment()

str

int

bool

list

dict

PrefectFuture

python
main_params = {
    "import_file": d_fut,   # not recommended
}

python
downloaded_path = d_fut.result()

main_params = {
    "import_file": str(downloaded_path),
    "working_dir": str(master_dir),
    "overwrite": overwrite,
}

ThreadPoolTaskRunner

DaskTaskRunner

python
with ThreadPoolTaskRunner(max_workers=8) as runner:
    download_futures = download_esa_ids.map(sorted_ids)

download_esa_ids.map(...)

FlowRunContext

DaskTaskRunner

with DaskTaskRunner(...)

task.submit()

task.map()

DaskTaskRunner

run_deployment(...)

ThreadPoolTaskRunner

DaskTaskRunner

download_futures

ThreadPoolTaskRunner

.map()

run_deployment

python
import asyncio
from pathlib import Path
from uuid import UUID

async def download_api_item_async(item: UUID) -> Path:
    # wrap your async I/O logic here
    ...

python
download_paths = await asyncio.gather(
    *(download_api_item_async(item) for item in sorted_ids)
)

download_paths

run_deployment(...)

ThreadPoolTaskRunner

Then the parent/coordinator flow can call this and get plain results back:

```python
@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": 50},
    ),
)
async def my_parent_coordinator_flow():
    sorted_ids: list[UUID] = ...

    downloaded_paths = download_items_flow(sorted_ids)

    submitted_runs = []
    for path in downloaded_paths:
        flow_run = await run_deployment(
            name="my-child-flow/child_deploy_shared_cluster",
            parameters={
                "import_file": str(path),
                "working_dir": str(master_dir),
                "overwrite": overwrite,
            },
            timeout=0,
            job_variables={
                "working_dir": str(create_clone_dir(working_dir=master_dir).name),
                "env": {
                    "DASK_SCHEDULER_ADDRESS": scheduler_address,
                },
            },
        )
        submitted_runs.append(flow_run)

asyncio.as_completed

python
async def download_api_item_async(item: UUID) -> Path:
    ...

python
async def submit_child_for_download(path: Path) -> dict:
    flow_run = await run_deployment(
        name="my-child-flow/child_deploy_shared_cluster",
        parameters={
            "import_file": str(path),
            "working_dir": str(master_dir),
            "overwrite": overwrite,
        },
        timeout=0,
        job_variables={
            "working_dir": str(create_clone_dir(working_dir=master_dir).name),
            "env": {"DASK_SCHEDULER_ADDRESS": scheduler_address},
        },
    )
    return {"flow_run_id": flow_run.id, "import_file": str(path)}

python
download_tasks = [asyncio.create_task(download_api_item_async(item)) for item in sorted_ids]

submitted_runs = []

for completed_download in asyncio.as_completed(download_tasks):
    path = await completed_download
    submitted = await submit_child_for_download(path)
    submitted_runs.append(submitted)

ThreadPoolTaskRunner

python
@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
def download_flow(...):
    ...

python
with ThreadPoolTaskRunner(...) as runner:
    my_task.map(...)

run_deployment()

ThreadPoolTaskRunner

DaskTaskRunner

run_deployment

ThreadPoolTaskRunner

ThreadPoolTaskRunner(max_workers=8)

DaskTaskRunner

download_flow

ThreadPoolTaskRunner

DaskTaskRunner

asyncio.as_completed(...)